{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "(spark-operator)=\n",
    "# Spark Operator runtime\n",
    "\n",
    "Using Spark Operator for running Spark jobs over k8s.<br>\n",
    "\n",
    "The `spark-on-k8s-operator` allows Spark applications to be defined in a declarative manner and supports one-time Spark \n",
    "applications with `SparkApplication` and cron-scheduled applications with `ScheduledSparkApplication`. <br>\n",
    "\n",
    "When sending a request with MLRun to the Spark operator, the request contains your full application configuration including the \n",
    "code and dependencies to run (packaged as a docker image or specified via URIs), the infrastructure parameters, (e.g. the \n",
    "memory, CPU, and storage volume specs to allocate to each Spark executor), and the Spark configuration.\n",
    "\n",
    "Kubernetes takes this request and starts the Spark driver in a Kubernetes pod (a k8s abstraction, just a docker container \n",
    "in this case). The Spark driver then communicates directly with the Kubernetes master to request executor pods, scaling them \n",
    "up and down at runtime according to the load if dynamic allocation is enabled. Kubernetes takes care of the bin-packing of \n",
    "the pods onto Kubernetes nodes (the physical VMs), and dynamically scales the various node pools to meet the requirements.\n",
    "\n",
    "When using Spark operator the resources are allocated per task, meaning that it scales down to zero when the task is done.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import mlrun\n",
    "import os\n",
    "\n",
    "# set up new spark function with spark operator\n",
    "# command will use our spark code which needs to be located on our file system\n",
    "# the name param can have only non capital letters (k8s convention)\n",
    "read_csv_filepath = os.path.join(os.path.abspath(\".\"), \"spark_read_csv.py\")\n",
    "sj = mlrun.new_function(kind=\"spark\", command=read_csv_filepath, name=\"sparkreadcsv\")\n",
    "\n",
    "# set spark driver config (gpu_type & gpus=<number_of_gpus>  supported too)\n",
    "sj.with_driver_limits(cpu=\"1300m\")\n",
    "sj.with_driver_requests(cpu=1, mem=\"512m\")\n",
    "\n",
    "# set spark executor config (gpu_type & gpus=<number_of_gpus> are supported too)\n",
    "sj.with_executor_limits(cpu=\"1400m\")\n",
    "sj.with_executor_requests(cpu=1, mem=\"512m\")\n",
    "\n",
    "# adds fuse, daemon & iguazio's jars support\n",
    "sj.with_igz_spark()\n",
    "\n",
    "# Alternately, move volume_mounts to driver and executor-specific fields and leave\n",
    "# v3io mounts out of executor mounts if mount_v3io_to_executor=False\n",
    "# sj.with_igz_spark(mount_v3io_to_executor=False)\n",
    "\n",
    "# set spark driver volume mount\n",
    "# sj.function.with_driver_host_path_volume(\"/host/path\", \"/mount/path\")\n",
    "\n",
    "# set spark executor volume mount\n",
    "# sj.function.with_executor_host_path_volume(\"/host/path\", \"/mount/path\")\n",
    "\n",
    "# confs are also supported\n",
    "sj.spec.spark_conf[\"spark.eventLog.enabled\"] = True\n",
    "\n",
    "# add python module\n",
    "sj.with_requiremants([`matplotlib`])\n",
    "\n",
    "# Number of executors\n",
    "sj.spec.replicas = 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Rebuilds the image with MLRun - needed in order to support logging artifacts etc.\n",
    "sj.deploy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run task while setting the artifact path on which the run artifact (in any) will be saved\n",
    "sj.run(artifact_path=\"/User\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "## Spark Code (spark_read_csv.py)\n",
    "\n",
    "```python\n",
    "from pyspark.sql import SparkSession\n",
    "from mlrun import get_or_create_ctx\n",
    "\n",
    "context = get_or_create_ctx(\"spark-function\")\n",
    "\n",
    "# build spark session\n",
    "spark = SparkSession.builder.appName(\"Spark job\").getOrCreate()\n",
    "\n",
    "# read csv\n",
    "df = spark.read.load('iris.csv', format=\"csv\",\n",
    "                     sep=\",\", header=\"true\")\n",
    "\n",
    "# sample for logging\n",
    "df_to_log = df.describe().toPandas()\n",
    "\n",
    "# log final report\n",
    "context.log_dataset(\"df_sample\",\n",
    "                     df=df_to_log,\n",
    "                     format=\"csv\")\n",
    "spark.stop()\n",
    "```"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
